package com.econ.powercloud.ems.service.common;

import com.econ.powercloud.clickhouse.dao.AnalogDataDAO;
import com.econ.powercloud.clickhouse.dao.StatusDataDAO;
import com.econ.powercloud.clickhouse.entity.AnalogData;
import com.econ.powercloud.clickhouse.entity.StatusData;
import com.econ.powercloud.ems.common.enums.DitTypeEnum;
import com.econ.powercloud.ems.common.enums.MeasureType;
import com.econ.powercloud.ems.constant.SysConstant;
import com.econ.powercloud.ems.dal.entity.EmsDevice;
import com.econ.powercloud.ems.pojo.DitDataVO;
import com.econ.powercloud.ems.pojo.DitHisDataVO;
import com.econ.powercloud.ems.pojo.HisDataVO;
import com.econ.powercloud.ems.pojo.scene.HisDataPidsVO;
import com.econ.powercloud.ems.pojo.scene.HisDataRequestVO;
import com.econ.powercloud.ems.pojo.scene.OpenApiHisDataVO;
import com.econ.powercloud.ems.service.dao.EmsDeviceDAO;
import com.econ.powercloud.ems.util.TimeUtil;
import com.econ.powercloud.ems.util.thread.AbstractMultiThread;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.function.Predicate;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * @Author : zjf
 * @Description : 查询历史数据工具类
 * @Date : 2021/12/9 10:41
 * @Modified by : zjf
 */
@Service
public class QueryHisDataCommonService {

    @Autowired
    private EmsDeviceDAO emsDeviceDAO;
    @Autowired
    private AnalogDataDAO analogDataDAO;
    @Autowired
    private StatusDataDAO statusDataDAO;

    /**
     * 查询某个设备的某些点号的某天数据
     * @param deviceCode：设备code
     * @param dits: 需要查询的点号后缀集合
     * @param time: 查询日期
     * @return key:pid
     */
    public Map<String, List<OpenApiHisDataVO>> queryYcHisDataByTime(String deviceCode, List<String> dits, Long time){
        Long startTime = TimeUtil.getStartOfDay(new Date(time));
        Long endTime = TimeUtil.getDayEndTime(new Date(time));
        EmsDevice emsDevice = emsDeviceDAO.selectByDeviceCode(deviceCode);
        List<List<String>> hisDataList = Lists.newArrayList();
        List<String> ditList = Lists.newArrayList();
        for (int i = 1; i <= dits.size(); i++){
            ditList.add(dits.get(i-1));
            if (!(i % 50 == 0 || i == dits.size())) {
                continue;
            }
            hisDataList.add(Lists.newArrayList(ditList));
            ditList.clear();
        }
        AbstractMultiThread<List<String>, Map<String,List<OpenApiHisDataVO>>> abstractMultiThread = new AbstractMultiThread<List<String>, Map<String,List<OpenApiHisDataVO>>>(hisDataList){
            @Override
            public Map<String,List<OpenApiHisDataVO>> outExecute(int currentThread, List<String> dits) {
                List<HisDataPidsVO> pids = Lists.newArrayList();
                pids.add(HisDataPidsVO.builder().deviceCode(emsDevice.getInitDeviceCode()).dits(dits).build());
                HisDataRequestVO hisDataRequestVO = HisDataRequestVO.builder()
                    .startTime(startTime).endTime(endTime).pids(pids)
                    .build();
                return getYcDataByListPid(hisDataRequestVO);
            }
        };
        List<Map<String,List<OpenApiHisDataVO>>> list = null;
        try {
            list = abstractMultiThread.getResult();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        Map<String, List<OpenApiHisDataVO>> result = Maps.newLinkedHashMap();
        if(CollectionUtils.isNotEmpty(list)){
            list.stream().forEach(d -> result.putAll(d));
        }
        return result;
    }

    public Map<String,List<OpenApiHisDataVO>> getYcDataByListPid(HisDataRequestVO hisDataRequestVO){
        if(CollectionUtils.isEmpty(hisDataRequestVO.getPids())){
            throw new IllegalArgumentException("传入的deviceCode或者点号不能为空！");
        }
        if (hisDataRequestVO.getStartTime() == null || hisDataRequestVO.getEndTime() == null) {
            throw new IllegalArgumentException("查询历史数据的开始时间与结束时间不能为空!");
        }
        if(hisDataRequestVO.getStartTime() > hisDataRequestVO.getEndTime()){
            throw new IllegalArgumentException("查询历史数据的开始时间不能大于结束时间!");
        }
        //存放查询历史数据的封装对象
        List<OpenApiHisDataVO> hisDataVOList = Lists.newArrayList();
        //pid集合
        List<String> pids = Lists.newArrayList();
        hisDataRequestVO.getPids().stream().forEach(hisAnologPidsVO->{
            hisAnologPidsVO.getDits().stream().forEach(dit -> pids.add(hisAnologPidsVO.getDeviceCode() + "." + dit));
        });
        List<AnalogData> analogDatas = analogDataDAO.selectByTimeInPid(hisDataRequestVO.getStartTime(), hisDataRequestVO.getEndTime(), pids);
        analogDatas.stream().forEach(d -> hisDataVOList.add(OpenApiHisDataVO.builder().id(d.getId()).pid(d.getPid()).time(d.getT()).v(d.getV()).build()));

        Map<String,List<OpenApiHisDataVO>> result = Maps.newLinkedHashMap();
        if(CollectionUtils.isNotEmpty(hisDataVOList)){
            hisDataVOList.stream().collect(Collectors.groupingBy(d->d.getPid())).forEach((k, v)->{
                result.put(k,v.stream().sorted(Comparator.comparing(OpenApiHisDataVO::getTime)).collect(Collectors.toList()));
            });
        }
        return result;
    }

    public Map<String,List<OpenApiHisDataVO>> getYxDataByListPid(HisDataRequestVO hisDataRequestVO){
        if(CollectionUtils.isEmpty(hisDataRequestVO.getPids())){
            throw new IllegalArgumentException("传入的deviceCode或者点号不能为空！");
        }
        if (hisDataRequestVO.getStartTime() == null || hisDataRequestVO.getEndTime() == null) {
            throw new IllegalArgumentException("查询历史数据的开始时间与结束时间不能为空!");
        }
        if(hisDataRequestVO.getStartTime() > hisDataRequestVO.getEndTime()){
            throw new IllegalArgumentException("查询历史数据的开始时间不能大于结束时间!");
        }
        //存放查询历史数据的封装对象
        List<OpenApiHisDataVO> hisDataVOList = Lists.newArrayList();
        //pid集合
        List<String> pids = Lists.newArrayList();
        hisDataRequestVO.getPids().stream().forEach(hisAnologPidsVO->{
            hisAnologPidsVO.getDits().stream().forEach(dit -> pids.add(hisAnologPidsVO.getDeviceCode() + "." + dit));
        });
        List<StatusData> statusDatas = statusDataDAO.selectByTimeInPid(hisDataRequestVO.getStartTime(), hisDataRequestVO.getEndTime(), pids);
        statusDatas.stream().forEach(d -> hisDataVOList.add(OpenApiHisDataVO.builder().id(d.getId()).pid(d.getPid()).time(d.getT()).v(d.getV()).build()));

        Map<String,List<OpenApiHisDataVO>> result = Maps.newLinkedHashMap();
        if(CollectionUtils.isNotEmpty(hisDataVOList)){
            hisDataVOList.stream().collect(Collectors.groupingBy(d->d.getPid())).forEach((k,v)->{
                result.put(k,v.stream().sorted(Comparator.comparing(OpenApiHisDataVO::getTime)).collect(Collectors.toList()));
            });
        }
        return result;
    }

    /**
     * 获取实时数据的点号的断面数据
     * @param sectionTime: 断面时间
     * @param hisDataVOList
     * @return
     */
    public List<OpenApiHisDataVO> selectFlashSectionData(Long sectionTime, List<HisDataVO> hisDataVOList){
        if(CollectionUtils.isEmpty(hisDataVOList)){
            return Collections.emptyList();
        }
        List<HisDataVO> ycDataList = hisDataVOList.stream().filter(d-> Objects.equals(MeasureType.yc.getCode(),d.getMeasureType())).collect(Collectors.toList());
        List<HisDataVO> yxDataList = hisDataVOList.stream().filter(d-> Objects.equals(MeasureType.yx.getCode(),d.getMeasureType())).collect(Collectors.toList());
        //查询遥测断面数据，并根据pid去重
        List<AnalogData> analogDataList = CollectionUtils.isEmpty(ycDataList)?Lists.newArrayList():analogDataDAO.selectSectionDataByPids(sectionTime,
                ycDataList.stream().map(HisDataVO::getPid).collect(Collectors.toList())).stream().filter(distinctByKey(d -> d.getPid())).collect(Collectors.toList());
        //查询遥信断面数据，并根据pid去重
        List<StatusData> statusDataList = CollectionUtils.isEmpty(yxDataList)?Lists.newArrayList():statusDataDAO.selectSectionDataByPids(sectionTime,
                yxDataList.stream().map(HisDataVO::getPid).collect(Collectors.toList())).stream().filter(distinctByKey(d -> d.getPid())).collect(Collectors.toList());
        //返回结果
        List<OpenApiHisDataVO> result = Lists.newArrayList();
        if(CollectionUtils.isNotEmpty(analogDataList)){
            analogDataList.stream().forEach(d -> result.add(OpenApiHisDataVO.builder().id(d.getId()).pid(d.getPid()).time(d.getT()).v(d.getV()).build()));
        }
        if(CollectionUtils.isNotEmpty(statusDataList)){
            statusDataList.stream().forEach(d -> result.add(OpenApiHisDataVO.builder().id(d.getId()).pid(d.getPid()).time(d.getT()).v(d.getV()).build()));
        }
        return result;
    }

    /**
     * 获取实时数据的点号的断面数据
     * @param startTime: 开始时间
     * @param endTime: 结束时间
     * @param hisDataVOList
     * @return
     */
    public List<OpenApiHisDataVO> selectFlashHisData(Long startTime,Long endTime, List<HisDataVO> hisDataVOList){
        if(CollectionUtils.isEmpty(hisDataVOList)){
            return Collections.emptyList();
        }
        List<HisDataVO> ycDataList = hisDataVOList.stream().filter(d-> Objects.equals(MeasureType.yc.getCode(),d.getMeasureType())).collect(Collectors.toList());
        List<HisDataVO> yxDataList = hisDataVOList.stream().filter(d-> Objects.equals(MeasureType.yx.getCode(),d.getMeasureType())).collect(Collectors.toList());
        //查询遥测断面数据，并根据pid去重
        List<AnalogData> analogDataList = CollectionUtils.isEmpty(ycDataList)?Lists.newArrayList():analogDataDAO.selectByTimeInPid(startTime,endTime,
                ycDataList.stream().map(HisDataVO::getPid).collect(Collectors.toList()));
        //查询遥信断面数据，并根据pid去重
        List<StatusData> statusDataList = CollectionUtils.isEmpty(yxDataList)?Lists.newArrayList():statusDataDAO.selectByTimeInPid(startTime,endTime,
                yxDataList.stream().map(HisDataVO::getPid).collect(Collectors.toList()));
        //返回结果
        List<OpenApiHisDataVO> result = Lists.newArrayList();
        if(CollectionUtils.isNotEmpty(analogDataList)){
            analogDataList.stream().forEach(d -> result.add(OpenApiHisDataVO.builder().id(d.getId()).pid(d.getPid()).time(d.getT()).v(d.getV()).build()));
        }
        if(CollectionUtils.isNotEmpty(statusDataList)){
            statusDataList.stream().forEach(d -> result.add(OpenApiHisDataVO.builder().id(d.getId()).pid(d.getPid()).time(d.getT()).v(d.getV()).build()));
        }
        return result;
    }

    /**
     * 获取点号数据的历史数据
     * @param startTime: 开始时间
     * @param endTime: 结束时间
     * @param hisDataVOList
     * @return
     */
    public List<OpenApiHisDataVO> selectHisDataByDitDataList(Long startTime, Long endTime, List<DitDataVO> hisDataVOList){
        if(CollectionUtils.isEmpty(hisDataVOList)){
            return Collections.emptyList();
        }
        List<DitDataVO> ycDataList = hisDataVOList.stream().filter(d-> DitTypeEnum.isYcData(d.getDitType())).collect(Collectors.toList());
        List<DitDataVO> yxDataList = hisDataVOList.stream().filter(d-> DitTypeEnum.isYxData(d.getDitType())).collect(Collectors.toList());
        //查询遥测历史数据
        List<AnalogData> analogDataList = CollectionUtils.isEmpty(ycDataList)?Lists.newArrayList():analogDataDAO.selectByTimeInPid(startTime,endTime,
                ycDataList.stream().map(d->d.getInitDeviceCode() + SysConstant.SPOT + d.getDit()).collect(Collectors.toList()));
        //查询遥信历史数据
        List<StatusData> statusDataList = CollectionUtils.isEmpty(yxDataList)?Lists.newArrayList():statusDataDAO.selectByTimeInPid(startTime,endTime,
                yxDataList.stream().map(d->d.getInitDeviceCode() + SysConstant.SPOT + d.getDit()).collect(Collectors.toList()));
        //返回结果
        List<OpenApiHisDataVO> result = Lists.newArrayList();
        if(CollectionUtils.isNotEmpty(analogDataList)){
            analogDataList.stream().forEach(d -> result.add(OpenApiHisDataVO.builder().id(d.getId()).pid(d.getPid()).time(d.getT()).v(d.getV()).build()));
        }
        if(CollectionUtils.isNotEmpty(statusDataList)){
            statusDataList.stream().forEach(d -> result.add(OpenApiHisDataVO.builder().id(d.getId()).pid(d.getPid()).time(d.getT()).v(d.getV()).build()));
        }
        return result;
    }

    /**
     * 函数式接口 T -> boolean
     * @param keyExtractor
     * @param <T>
     * @return
     */
    public static <T> Predicate<T> distinctByKey(Function<? super T, ?> keyExtractor) {
        ConcurrentHashMap<Object, Boolean> map = new ConcurrentHashMap<>(16);
        return t -> map.putIfAbsent(keyExtractor.apply(t),Boolean.TRUE) == null;
    }
}
